package org.apache.hadoop.hdfs.server.namenode.ha;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.security.PrivilegedAction;
import java.security.PrivilegedExceptionAction;
import java.util.Collection;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.HAUtil;
import org.apache.hadoop.hdfs.protocol.HdfsConstants;
import org.apache.hadoop.hdfs.protocolPB.NamenodeProtocolPB;
import org.apache.hadoop.hdfs.protocolPB.NamenodeProtocolTranslatorPB;
import org.apache.hadoop.hdfs.server.namenode.EditLogInputException;
import org.apache.hadoop.hdfs.server.namenode.EditLogInputStream;
import org.apache.hadoop.hdfs.server.namenode.FSEditLog;
import org.apache.hadoop.hdfs.server.namenode.FSImage;
import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
import org.apache.hadoop.hdfs.server.namenode.NameNode;
import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocol;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.security.SecurityUtil;

import static org.apache.hadoop.util.Time.now;
import static org.apache.hadoop.util.ExitUtil.terminate;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;

/**
 * EditLogTailer代表了一个后台线程，这个后台线会不断的周期性的从journal nodes集群上拉取edits log数据流，
 * 接着将edits log数据流应用到自己的 FSNamesystem 上去，也就是自己本地的元数据上去。
 */
@InterfaceAudience.Private
@InterfaceStability.Evolving
public class EditLogTailer {
  public static final Log LOG = LogFactory.getLog(EditLogTailer.class);

  /** 编辑日志跟踪线程EditLogTailerThread实例tailerThread */
  private final EditLogTailerThread tailerThread;

  /** HDFS配置信息Configuration实例conf */
  private final Configuration conf;
  /** 文件系统命名空间FSNamesystem实例namesystem */
  private final FSNamesystem namesystem;
  /** 文件系统编辑日志FSEditLog实例editLog */
  private FSEditLog editLog;

  /** Active NameNode地址InetSocketAddress */
  private InetSocketAddress activeAddr;
  /** 名字节点通信接口NamenodeProtocol */
  private NamenodeProtocol cachedActiveProxy = null;

  /** 一次编辑日志滚动开始时的最新事务ID */
  private long lastRollTriggerTxId = HdfsConstants.INVALID_TXID;
  
  /** StandBy NameNode加载的最高事务ID */
  private long lastLoadedTxnId = HdfsConstants.INVALID_TXID;

  /** 最后一次我们从共享目录成功加载一个非零编辑的时间 */
  private long lastLoadTimestamp;

  /** StandBy NameNode滚动编辑日志的时间间隔 */
  private final long logRollPeriodMs;

  /** StandBy NameNode检查是否存在可以读取的新的最终日志段的时间间隔 */
  private final long sleepTimeMs;
  
  public EditLogTailer(FSNamesystem namesystem, Configuration conf) {
    this.tailerThread = new EditLogTailerThread(); // 实例化编辑日志追踪线程EditLogTailerThread
    this.conf = conf; // 根据入参初始化配置信息conf和文件系统命名系统namesystem
    this.namesystem = namesystem;
    this.editLog = namesystem.getEditLog(); // 从namesystem中获取editLog
    
    lastLoadTimestamp = now(); // 最新加载edit log时间lastLoadTimestamp初始化为当前时间

    // 取参数dfs.ha.log-roll.period，参数未配置默认为2min
    logRollPeriodMs = conf.getInt(DFSConfigKeys.DFS_HA_LOGROLL_PERIOD_KEY, DFSConfigKeys.DFS_HA_LOGROLL_PERIOD_DEFAULT) * 1000;

    // 如果logRollPeriodMs大于等于0
    if (logRollPeriodMs >= 0) {
      // 调用getActiveNodeAddress()方法初始化Active NameNode地址activeAddr
      this.activeAddr = getActiveNodeAddress();
      Preconditions.checkArgument(activeAddr.getPort() > 0, "Active NameNode must have an IPC port configured. " + "Got address '%s'", activeAddr);
      LOG.info("Will roll logs on active node at " + activeAddr + " every " + (logRollPeriodMs / 1000) + " seconds.");
    } else {
      LOG.info("Not going to trigger log rolls on active node because " + DFSConfigKeys.DFS_HA_LOGROLL_PERIOD_KEY + " is negative.");
    }

    // 获取参数dfs.ha.tail-edits.period，参数未配置默认为1min
    sleepTimeMs = conf.getInt(DFSConfigKeys.DFS_HA_TAILEDITS_PERIOD_KEY, DFSConfigKeys.DFS_HA_TAILEDITS_PERIOD_DEFAULT) * 1000;
    
    LOG.debug("logRollPeriodMs=" + logRollPeriodMs +  " sleepTime=" + sleepTimeMs);
  }
  
  private InetSocketAddress getActiveNodeAddress() {
    Configuration activeConf = HAUtil.getConfForOtherNode(conf);
    return NameNode.getServiceAddress(activeConf, true);
  }
  
  private NamenodeProtocol getActiveNodeProxy() throws IOException {
    if (cachedActiveProxy == null) {
      int rpcTimeout = conf.getInt(
          DFSConfigKeys.DFS_HA_LOGROLL_RPC_TIMEOUT_KEY,
          DFSConfigKeys.DFS_HA_LOGROLL_RPC_TIMEOUT_DEFAULT);
      NamenodeProtocolPB proxy = RPC.waitForProxy(NamenodeProtocolPB.class,
          RPC.getProtocolVersion(NamenodeProtocolPB.class), activeAddr, conf,
          rpcTimeout, Long.MAX_VALUE);
      cachedActiveProxy = new NamenodeProtocolTranslatorPB(proxy);
    }
    assert cachedActiveProxy != null;
    return cachedActiveProxy;
  }

  public void start() {
    tailerThread.start();
  }
  
  public void stop() throws IOException {
    tailerThread.setShouldRun(false);
    tailerThread.interrupt();
    try {
      tailerThread.join();
    } catch (InterruptedException e) {
      LOG.warn("Edit log tailer thread exited with an exception");
      throw new IOException(e);
    }
  }
  
  @VisibleForTesting
  FSEditLog getEditLog() {
    return editLog;
  }
  
  @VisibleForTesting
  public void setEditLog(FSEditLog editLog) {
    this.editLog = editLog;
  }
  
  public void catchupDuringFailover() throws IOException {
    Preconditions.checkState(tailerThread == null ||
        !tailerThread.isAlive(),
        "Tailer thread should not be running once failover starts");
    // Important to do tailing as the login user, in case the shared
    // edits storage is implemented by a JournalManager that depends
    // on security credentials to access the logs (eg QuorumJournalManager).
    SecurityUtil.doAsLoginUser(new PrivilegedExceptionAction<Void>() {
      @Override
      public Void run() throws Exception {
        try {
          doTailEdits();
        } catch (InterruptedException e) {
          throw new IOException(e);
        }
        return null;
      }
    });
  }

  /**
   * 大体处理流程如下：
   * 1、首先，namesystem加写锁；
   * 2、通过namesystem获取文件系统镜像FSImage实例image；
   * 3、通过文件系统镜像FSImage实例image获取最新的事务ID，即lastTxnId；
   * 4、从编辑日志editLog中获取编辑日志输入流集合streams，获取的输入流为最新事务ID加1之后的数据：
   *    ps：注意，这个编辑日志输入流集合streams并非读取的是editLog对象中的数据，毕竟editLog也是根据namesystem来获取的，如果从其中读取数据再加载到namesystem中的fsimage中，没有多大意义，
   *    这个日志输入流实际上是通过Hadoop HA中的JournalNode来获取的，。
   * 5、调用文件系统镜像FSImage实例image的loadEdits()，利用编辑日志输入流集合streams，加载编辑日志至目标namesystem中的文件系统镜像FSImage，并获得编辑日志加载的大小editsLoaded；
   * 6、如果editsLoaded大于0，最后一次我们从共享目录成功加载一个非零编辑的时间lastLoadTimestamp更新为当前时间；
   * 7、上次StandBy NameNode加载的最高事务ID更新为image中最新事务ID；
   * 8、namesystem去除写锁。
   */
  @VisibleForTesting
  void doTailEdits() throws IOException, InterruptedException {
    // namesystem加写锁
    namesystem.writeLockInterruptibly();
    try {
      // 通过namesystem获取文件系统镜像FSImage实例image
      FSImage image = namesystem.getFSImage();
      // 通过文件系统镜像FSImage实例image获取最新的事务ID,即standby namenode本地元数据上的eidts log的txid
      long lastTxnId = image.getLastAppliedTxId();
      
      Collection<EditLogInputStream> streams;
      try {
        // 从编辑日志editLog中获取编辑日志输入流集合streams，获取的输入流为lastTxnId+1之后的edits log
        streams = editLog.selectInputStreams(lastTxnId + 1, 0, null, false);
      } catch (IOException ioe) {
        LOG.warn("Edits tailer failed to find any streams. Will try again " + "later.", ioe);
        return;
      }

      long editsLoaded = 0;
      try {
        // 调用文件系统镜像FSImage实例image的loadEdits()，
        // 利用编辑日志输入流集合streams，加载编辑日志至目标namesystem中的文件系统镜像FSImage， 并获得编辑日志加载的大小editsLoaded
        // FSImage代表了人家的元数据，底层使用了EditLogInputStream 从journalnode尝试拉取更新的edits log，
        // 如果拉取到了，肯定在这个方法的内部，就会应用到元数据里去
        editsLoaded = image.loadEdits(streams, namesystem); //
      } catch (EditLogInputException elie) {
        editsLoaded = elie.getNumEditsLoaded();
        throw elie;
      } finally {
        if (editsLoaded > 0 || LOG.isDebugEnabled()) {
          LOG.info(String.format("Loaded %d edits starting from txid %d ", editsLoaded, lastTxnId));
        }
      }

      if (editsLoaded > 0) { // 如果editsLoaded大于0
        // 最后一次我们从共享目录成功加载一个非零编辑的时间lastLoadTimestamp更新为当前时间
        lastLoadTimestamp = now();
      }
      // 上次StandBy NameNode加载的最高事务ID更新为image中最新事务ID
      lastLoadedTxnId = image.getLastAppliedTxId();
    } finally {
      // namesystem去除写锁
      namesystem.writeUnlock();
    }
  }

  /**
   * @return timestamp (in msec) of when we last loaded a non-zero number of edits.
   */
  public long getLastLoadTimestamp() {
    return lastLoadTimestamp;
  }

  /**
   *  standBy NameNode滚动edits log日志的时间间隔logRollPeriodMs大于0，
   * 且最后一次我们从共享目录成功加载一个非零编辑的时间到现在的时间间隔大于logRollPeriodMs
   */
  private boolean tooLongSinceLastLoad() {
    return logRollPeriodMs >= 0 &&  (now() - lastLoadTimestamp) > logRollPeriodMs ;
  }

  /**
   * Trigger the active node to roll its logs.
   */
  private void triggerActiveLogRoll() {
    LOG.info("Triggering log roll on remote NameNode " + activeAddr);
    try {
      // 获得Active NameNode的代理，并调用其rollEditLog()方法滚动编辑日志
      getActiveNodeProxy().rollEditLog();
      // 将上次StandBy NameNode加载的最高事务ID，即lastLoadedTxnId，赋值给上次edits log滚动开始时的最新事务ID，即lastRollTriggerTxId，
      // 这么做是为了方便进行日志回滚
      lastRollTriggerTxId = lastLoadedTxnId;
    } catch (IOException ioe) {
      LOG.warn("Unable to trigger a roll of the active NN", ioe);
    }
  }

  /**
   * 执行跟踪编辑日志并将事务应用到 FSNS 的实际工作的线程。
   * 编辑日志跟踪线程，
   */
  private class EditLogTailerThread extends Thread {
    private volatile boolean shouldRun = true;
    
    private EditLogTailerThread() {
      super("Edit log tailer");
    }
    
    private void setShouldRun(boolean shouldRun) {
      this.shouldRun = shouldRun;
    }
    
    @Override
    public void run() {
      SecurityUtil.doAsLoginUserOrFatal(
          new PrivilegedAction<Object>() {
          @Override
          public Object run() {
            doWork();
            return null;
          }
        });
    }

    /**
     * 当标志位shouldRun为true时，doWork()方法一直在while循环内执行，其处理逻辑如下：
     * 1、如果是自从上次加载后过了太长时间，并且上次编辑日志滚动开始时的最新事务ID小于上次StandBy NameNode加载的最高事务ID，触发Active NameNode进行编辑日志滚动：
     *     自从上次加载后过了太长时间是根据tooLongSinceLastLoad()方法判断的，而触发Active NameNode进行编辑日志滚动则是通过triggerActiveLogRoll()方法来完成的；
     * 2、判断标志位shouldRun，如果其为false的话，退出循环；
     * 3、调用doTailEdits()方法执行日志追踪；
     * 4、线程休眠sleepTimeMs时间后继续执行上述工作。
     */
    private void doWork() {
      // 标志位shouldRun为true时一直循环
      while (shouldRun) {
        try {
          // 自从上次日志滚动触发以来，如果StandBy NameNode没有读到任何事务的话，触发一次日志滚动，
          // 如果是自从上次加载后过了太长时间，并且上次edits log滚动开始时的最新事务ID小于上次StandBy NameNode加载的最高事务ID

          // edits log roll，重新创建一个新的edits_inprogress文件
          // 之前的文件就固定为 startTransactionId-endTransactionId.log
          if (tooLongSinceLastLoad() && lastRollTriggerTxId < lastLoadedTxnId) {
            // 触发Active NameNode进行edits log日志滚动
            triggerActiveLogRoll();
          }

          // 判断标志位shouldRun，如果其为false的话，退出循环
          if (!shouldRun) {
            break;
          }

          // 核心代码：调用doTailEdits()方法执行日志追踪
          doTailEdits(); //
        } catch (EditLogInputException elie) {
          LOG.warn("Error while reading edits from disk. Will try again.", elie);
        } catch (InterruptedException ie) {
          continue;
        } catch (Throwable t) {
          LOG.fatal("Unknown error encountered while tailing edits. " +  "Shutting down standby NN.", t);
          terminate(1, t);
        }

        try {
          // standby namenode 默认每个 1分钟 尝试获取 journalnode上的edits log
          Thread.sleep(sleepTimeMs); // 60s
        } catch (InterruptedException e) {
          LOG.warn("Edit log tailer interrupted", e);
        }
      }
    }
  }

}
